package com.riven.redisson.message;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.riven.redisson.util.JSON;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.Decoder;
import org.redisson.client.protocol.Encoder;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author riven
 */
public class RedissonMessageCodec extends StringCodec {

    public static final RedissonMessageCodec INSTANCE = new RedissonMessageCodec();

    private final Encoder encoder;

    private final Decoder<Object> decoder;

    public RedissonMessageCodec() {
        this(StandardCharsets.UTF_8);
    }

    public RedissonMessageCodec(Charset charset) {
        super(charset);
        this.encoder = object -> {
            String jsonStr = JSON.toJSONString(object);
            return super.getValueEncoder().encode(jsonStr);
        };
        this.decoder = (buf, state) -> {
            byte[] result = new byte[buf.readableBytes()];
            buf.readBytes(result);
            JsonNode jsonNode = JSON.OBJECT_MAPPER.readTree(result);
            String payload = jsonNode.get("payload").toString();
            Map<String, Object> headers = null;
            JsonNode headJson;
            if ((headJson = jsonNode.get("headers")) != null) {
                headers = JSON.OBJECT_MAPPER.readValue(headJson.toString(), new TypeReference<>() {
                });
            }
            return new RedissonMessage(payload == null ? new byte[]{} : payload.getBytes(charset), headers);
        };
    }

    @Override
    public Decoder<Object> getValueDecoder() {
        return this.decoder;
    }

    @Override
    public Encoder getValueEncoder() {
        return this.encoder;
    }

}
